Luna Model

A Luna model is a custom data type which can be used within Template definitions at runtime for:

• Event payload structures
• Template Parameters
• Template Variables
• general use in Clause logic

It is most simple to think of LunaModel as a pydantic BaseModel (because this is what is extended).

A LunaModel can be nested into others if desired.

Below is an example of a Template which uses Luna Models to manage simple market data, with some light randomization.

import random
from decimal import Decimal

import luna
from luna.types import LunaModel

class Tag(LunaModel):
    name: str
    dt: datetime

class MktData(LunaModel):
    stock: str
    price: Decimal
    price_time: datetime

    tags: list[Tag]

class Currency(LunaModel):
    """
    Represents a currency with its name, code, and exchange rate to USD.

    Attributes:
    - `name` (str): The name of the currency (e.g., "US dollar").
    - `code` (str): The currency code (e.g., "USD").
    - `usd_exchange_rate` (Decimal): The exchange rate of this currency to USD.
    """

    name: str
    code: str
    usd_exchange_rate: Decimal  # how much to get one usd


class Security(LunaModel):
    """
    Represents a financial security with its reference, base currency, and price.

    Attributes:
    - `ref` (str): The reference identifier for the security (e.g., "MSFT").
    - `base_currency` (str): The base currency of the security (e.g., "USD").
    - `price` (Decimal): The price of the security.
    """

    ref: str
    base_currency: str
    price: Decimal  # how much to get one usd

class MarketData(luna.Template):
    """
    This template models market data management, including the dynamic handling of currency and security data.
    Features include:
    - Definition of currency and security data structures using LunaModel.
    - Dynamic updating of currency exchange rates and security prices at regular intervals.
    - Event-driven methods for setting up initial data for currencies and securities.
    """

    broker = luna.Role()

    currencies = luna.Variable(type_=list[Currency], value=[])
    securities = luna.Variable(type_=list[Security], value=[])

    @luna.clause(
        on=luna.Event(name="setup_currency", payload={"value": list[Currency]})
    )
    def create_currencies(self, ctx, event):
        """
        Creates or updates the list of currencies based on the provided event payload.

        **Parameters:**
        - `ctx` (Context): Provides access to the current state, environment, and utilities of the Luna execution context.
        - `event` (Event): The event containing the currency data to be set up or updated.
        """

        res = [x for x in self.currencies]
        for currency in event.payload.value:
            idx, _ = next(
                ((i, c) for (i, c) in enumerate(res) if c.code == currency.code),
                (None, None),
            )
            if idx:
                del res[idx]

            res += [currency]

        self.currencies = res

    @luna.clause(
        on=luna.Event(name="setup_securities", payload={"value": list[Security]})
    )
    def create_securities(self, ctx, event):
        """
        Creates or updates the list of securities based on the provided event payload.

        **Parameters:**
        - `ctx` (Context): Provides access to the current state, environment, and utilities of the Luna execution context.
        - `event` (Event): The event containing the security data to be set up or updated.
        """
        res = [x for x in self.securities]
        for security in event.payload.value:
            idx, _ = next(
                (
                    (i, c)
                    for (i, c) in enumerate(self.securities)
                    if c.ref == security.ref
                ),
                (None, None),
            )
            if idx:
                del res[idx]

            res += [security]

        self.securities = res

    @luna.clause(on=luna.Every("30 minutes"))
    def update_currency(self, ctx):
        """
        Periodically updates the exchange rates of the currencies.

        **Parameters:**
        - `ctx` (Context): Provides access to the current state, environment, and utilities of the Luna execution context.
        """
        res = []
        for currency in self.currencies:
            if currency.code.lower() == "usd":
                res.append(currency)
                continue

            r = Decimal(random.randrange(1, 200)) / 10000
            r *= random.randint(-1, 1)
            ccy_amt = Decimal(currency.usd_exchange_rate) * (1 + r)

            res.append(
                Currency(
                    name=currency.name,
                    code=currency.code,
                    usd_exchange_rate=ccy_amt,
                )
            )

        self.currencies = res

    @luna.clause(on=luna.Every("30 minutes"))
    def update_security(self, ctx):
        """
        Periodically updates the prices of the securities.

        **Parameters:**
        - `ctx` (Context): Provides access to the current state, environment, and utilities of the Luna execution context.
        """
        res = []
        for security in self.securities:
            r = Decimal(random.randrange(1, 200)) / 10000
            r *= random.choice([-1, 1])
            s = Decimal(security.price) * (1 + r)

            res.append(
                Security(
                    ref=security.ref,
                    base_currency=security.base_currency,
                    price=s,
                )
            )

        self.securities = res